package stc.skymobi.netty.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.util.AttributeKey;

import org.apache.commons.pool2.ObjectPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import stc.skymobi.bean.tlv.annotation.HttpTLVAttribute;
import stc.skymobi.netty.handler.codec.http.response.DefaultIdentifiableHttpResponse;
import stc.skymobi.netty.transport.PoolableSender;
import stc.skymobi.netty.util.NettyConstants;
import stc.skymobi.util.Identifyable;
import stc.skymobi.util.MutableIdentifyable;

import java.util.UUID;

public class HttpClient implements PoolableSender {
    private static Logger logger = LoggerFactory.getLogger(HttpClient.class);

    private static final NioEventLoopGroup NIO_EVENT_LOOP_GROUP = new NioEventLoopGroup();

    private static final AttributeKey<UUID> CTX_UUID = AttributeKey.valueOf("CTX_UUID");

    private ChannelInitializer httpClientInitializer;
    private Bootstrap bootstrap = new Bootstrap();
    private String remoteAddress;
    private int remotePort;
    private String localAddress = "0.0.0.0";
    private int localPort;
    private int connectTimeoutMillis = 10000; // 10 seconds

    private boolean connected;
    private Channel channel;
    private boolean isReturnedToPool;

    private ObjectPool pool;
    private HttpClient client = this;

    private final class ChannelObserver extends ChannelInboundHandlerAdapter {

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            returnClientToPool("messageReceived");
            UUID uuid = ctx.channel().attr(CTX_UUID).get();
            if (uuid != null) {
                ctx.channel().attr(CTX_UUID).remove();
                if (msg instanceof FullHttpResponse) {
                    DefaultIdentifiableHttpResponse response = new DefaultIdentifiableHttpResponse();
                    response.setUuid(uuid);
                    response.setHttpResponse((FullHttpResponse) msg);
                    super.channelRead(ctx, response);
                    return;
                } else if (msg instanceof MutableIdentifyable) {
                    ((MutableIdentifyable) msg).setIdentification(uuid);
                }
            }

            super.channelRead(ctx, msg);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            connected = false;
            channel = null;
            returnClientToPool("channelInactive");
            super.channelInactive(ctx);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            logger.error("exceptionCaught ", cause);
            returnClientToPool("exceptionCaught");
            super.exceptionCaught(ctx, cause);
        }

        @SuppressWarnings("unchecked")
        private void returnClientToPool(String reason) throws Exception {
            if (pool == null) {
                logger.info("http client pool is not set,there will be pool leak if you are using a http client pool.");
            } else if (client.isReturnedToPool()) {
                logger.debug("{},client {} is returned to pool already, so don't return it again", reason, client);
            } else {
                client.setReturnedToPool(true);
                pool.returnObject(client);
                logger.debug("{},and returned {} to pool", reason, client);
            }
        }
    }

    public void send(Object message) {
        if (message == null)
            return;
        if (message instanceof Identifyable) {   //在Channel中加上uuid
            channel.attr(CTX_UUID).set(((Identifyable) message).getIdentification());
        }
        HttpTLVAttribute attr = message.getClass().getAnnotation(HttpTLVAttribute.class);
        if (attr != null) {
            channel.attr(NettyConstants.CTX_HTTP_RESP_CLASS).set(attr.responseType());
        }
        channel.writeAndFlush(message);
    }

    public HttpClient connect() throws Exception {
        bootstrap.group(NIO_EVENT_LOOP_GROUP)
                .channel(NioSocketChannel.class)
                .localAddress(localAddress, localPort)
                .remoteAddress(remoteAddress, remotePort)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMillis)
                .handler(httpClientInitializer);

        ChannelFuture future = bootstrap.connect();
        logger.debug("connection opened,waiting for channel connected!");
        channel = future.await().channel();
        if (channel.pipeline().get("handler") != null) {//确保在将http应答提交给业务层之前将uuid装配好
            channel.pipeline().addBefore("handler", "observer", new ChannelObserver());
        } else {
            channel.pipeline().addLast("observer", new ChannelObserver());
        }

        logger.debug("new http client channel established!");

        return this;
    }

    public void close() {
        if (channel != null) {
            channel.close();
        }
    }

    @Override
    public String toString() {
        return "httpClient[" + channel + "]";
    }

    public String getRemoteAddress() {
        return remoteAddress;
    }

    public void setRemoteAddress(String remoteAddress) {
        this.remoteAddress = remoteAddress;
    }

    public int getRemotePort() {
        return remotePort;
    }

    public void setRemotePort(int remotePort) {
        this.remotePort = remotePort;
    }

    public boolean isConnected() {
        return connected;
    }

    public Channel getChannel() {
        return channel;
    }

    public String getLocalAddress() {
        return localAddress;
    }

    public void setLocalAddress(String localAddress) {
        this.localAddress = localAddress;
    }

    public int getLocalPort() {
        return localPort;
    }

    public void setLocalPort(int localPort) {
        this.localPort = localPort;
    }

    public void setPool(ObjectPool pool) {
        this.pool = pool;
    }

    public void setHttpClientInitializer(ChannelInitializer httpClientInitializer) {
        this.httpClientInitializer = httpClientInitializer;
    }

    public boolean isReturnedToPool() {
        return isReturnedToPool;
    }

    public void setReturnedToPool(boolean returnedToPool) {
        isReturnedToPool = returnedToPool;
    }

    public void setConnectTimeoutMillis(int connectTimeoutMillis) {
        this.connectTimeoutMillis = connectTimeoutMillis;
    }
}
